/*
 *    Copyright 2012 The MyBatis Team
 *
 *    Licensed under the Apache License, Version 2.0 (the "License");
 *    you may not use this file except in compliance with the License.
 *    You may obtain a copy of the License at
 *
 *       http://www.apache.org/licenses/LICENSE-2.0
 *
 *    Unless required by applicable law or agreed to in writing, software
 *    distributed under the License is distributed on an "AS IS" BASIS,
 *    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *    See the License for the specific language governing permissions and
 *    limitations under the License.
 */
package org.apache.ibatis.caches.memcached;

import java.io.Serializable;
import java.util.HashSet;
import java.util.Set;

import net.rubyeye.xmemcached.MemcachedClient;

import org.apache.ibatis.cache.CacheException;
import org.apache.ibatis.logging.Log;
import org.apache.ibatis.logging.LogFactory;

/**
 * 
 * @author chenld(liangdengchen@cyou-inc.com)
 *
 */
public final class XMemcachedClientWrapper {

    /**
     * This class log.
     */
    private final Log       log       = LogFactory.getLog(XMemcachedCache.class);

    private String          keyPrefix = "mybatis-xmemcached-";
    
    private static final String KEY_STRING_FIX = "cache-sqlkey-md5:";       
    
    private static final String KEY_GROUP_FIX = "cache-groupkey-";   

    /**
     * unit:seconds
     */
    private int             expiration;

    /**
     * memcached client
     */
    private MemcachedClient memcachedClient;

    public XMemcachedClientWrapper(){
        MybatisXMemcachedClientBuilder memcachedClientBuilder = (MybatisXMemcachedClientBuilder) SpringContextHolder.getApplicationContext().getBean("memcachedClientBuilder",
                                                                                                                                                     MybatisXMemcachedClientBuilder.class);
        memcachedClient = (MemcachedClient) SpringContextHolder.getApplicationContext().getBean("xmemcachedClient",
                                                                                                MemcachedClient.class);
        keyPrefix = memcachedClientBuilder.getKeyPrefix();
        expiration = memcachedClientBuilder.getExpiration();
    }

    /**
     * Converts the iBatis object key in the proper string representation.
     * 
     * @param key the iBatis object key.
     * @return the proper string representation.
     */
    private String toKeyString(final Object key) {
        String _key = KEY_STRING_FIX + MD5Util.getMD5String(key.toString());
        String keyString = keyPrefix + _key; 
        
        if (log.isDebugEnabled()) {
            log.debug("Object key '" + key + "' converted in '" + keyString + "'");
        }
        return keyString;
    }
    

    /**
     * Converts the iBatis object key in the proper string representation.
     * 
     * @param key the iBatis object key.
     * @return the proper string representation.
     */
    private String toGroupKeyString(final Object key) {
        String keyString = keyPrefix + KEY_GROUP_FIX + key; 
        
        if (log.isDebugEnabled()) {
            log.debug("Object key '" + key + "' converted in '" + keyString + "'");
        }
        return keyString;
    }

    /**
     * @param key
     * @return
     */
    public Object getObject(Object key, String id) {
        String keyString = toKeyString(key);
        String groupKey = toGroupKeyString(id);

        Object ret = retrieve(keyString);

        if (log.isDebugEnabled()) {
            log.debug("Retrived object (" + keyString + ", " + ret + ")");
        }

        storeGroupInMemcached(groupKey, keyString);

        return ret;
    }

    /**
     * Return the stored group in Memcached identified by the specified key.
     * 
     * @param groupKey the group key.
     * @return the group if was previously stored, null otherwise.
     */
    @SuppressWarnings("unchecked")
    private Set<String> getGroup(String groupKey) {
        if (log.isDebugEnabled()) {
            log.debug("Retrieving group with id '" + groupKey + "'");
        }

        Object groups = null;
        try {
            groups = retrieve(groupKey);
        } catch (Exception e) {
            log.error("Impossible to retrieve group '" + groupKey + "' see nested exceptions", e);
        }

        if (groups == null) {
            if (log.isDebugEnabled()) {
                log.debug("Group '" + groupKey + "' not previously stored");
            }
            return null;
        }

        if (log.isDebugEnabled()) {
            log.debug("retrieved group '" + groupKey + "' with values " + groups);
        }
        return (Set<String>) groups;
    }

    /**
     * @param keyString
     * @return
     * @throws Exception
     */
    private Object retrieve(final String keyString) {
        Object retrieved = null;
        try {
            retrieved = memcachedClient.get(keyString);
        } catch (Exception e) {
            e.printStackTrace();
        }
        return retrieved;
    }

    public void putObject(Object key, Object value, String id) {
        String keyString = toKeyString(key);
        String groupKey = toGroupKeyString(id);

        if (log.isDebugEnabled()) {
            log.debug("Putting object (" + keyString + ", " + value + ")");
        }

        storeInMemcached(keyString, value);

        storeGroupInMemcached(groupKey, keyString);
    }

    private void storeGroupInMemcached(String groupKey, String keyString) {
        // add namespace key into memcached
        boolean dirty = false;

        Set<String> group = getGroup(groupKey);
        if (group == null) {
            group = new HashSet<String>();
            dirty = true;
        }

        if (!group.contains(keyString)) {
            group.add(keyString);
            dirty = true;
        }

        if (log.isDebugEnabled()) {
            log.debug("Insert/Updating object (" + groupKey + ", " + group + ")");
        }

        if (dirty) {
            storeInMemcached(groupKey, group);
        }

    }

    private void deleteGroupInMemcached(String groupKey, String keyString) {
        // add namespace key into memcached
        boolean dirty = false;
        Set<String> group = getGroup(groupKey);

        if (group == null) {
            return;
        }

        if (group.contains(keyString)) {
            group.remove(keyString);
            dirty = true;
        }

        if (log.isDebugEnabled()) {
            log.debug("remove object (" + groupKey + ", " + group + ")");
        }

        if (dirty) {
            storeInMemcached(groupKey, group);
        }
    }

    /**
     * Stores an object identified by a key in Memcached.
     * 
     * @param keyString the object key
     * @param value the object has to be stored.
     */
    private void storeInMemcached(String keyString, Object value) {
        if (value != null && !Serializable.class.isAssignableFrom(value.getClass())) {
            throw new CacheException("Object of type '" + value.getClass().getName()
                                     + "' that's non-serializable is not supported by Memcached");
        }
        try {
            memcachedClient.set(keyString, expiration, value);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }

    public Object removeObject(Object key, String id) {
        String keyString = toKeyString(key);
        String groupKey = toGroupKeyString(id);

        if (log.isDebugEnabled()) {
            log.debug("Removing object '" + keyString + "'");
        }

        Object result = getObject(key, id);
        if (result != null) {
            try {
                memcachedClient.delete(keyString);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        deleteGroupInMemcached(groupKey, keyString);

        return result;
    }

    public void removeGroup(String id) {
        String groupKey = toGroupKeyString(id);

        Set<String> group = getGroup(groupKey);

        if (group == null) {
            if (log.isDebugEnabled()) {
                log.debug("No need to flush cached entries for group '" + id + "' because is empty");
            }
            return;
        }

        if (log.isDebugEnabled()) {
            log.debug("Flushing keys: " + group);
        }

        for (String key : group) {
            try {
                memcachedClient.delete(key);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }

        if (log.isDebugEnabled()) {
            log.debug("Flushing group: " + groupKey);
        }

        try {
            memcachedClient.delete(groupKey);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    protected void finalize() throws Throwable {
        super.finalize();
        memcachedClient.shutdown();
    }

}
